package com.wzh.project.kafka.util;

import java.util.Collections;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Future;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.wzh.project.common.Constants;


/**
 * kafka操作工具类
 * @author wzh
 *
 */
public class KafkaUtil {
	private static final Logger logger = LogManager.getLogger(KafkaUtil.class);
	public static void main(String[] args) throws Exception {
//		new Thread().start();
		reciveMessage();
//		sendMessage("");
	}
	
	
	public static void reciveMessage() throws Exception{
		System.out.println("reciveMessage");
		Properties p = new Properties();
//		p.put("zookeeper.connect", "10.23.11.132:2181");
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.23.11.132:9092");
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "group12");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
        kafkaConsumer.subscribe(Collections.singletonList(Constants.KAFKA_TOPIC));// 订阅消息

        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(String.format("topic:%s,offset:%d,消息:%s", //
                        record.topic(), record.offset(), record.value()));
            }
        }
	}
}
